package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.SimpleDateFormat;
import java.util.Date;

public class UnVisitAppdd {
    public static void main(String[] args) throws Exception {
        //设置flink的处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度
        env.setParallelism(4);

        //在kafka中读取数据
        String sourceTopic="dwd_page_log";
        String groupId="unique_visit_app_group";
        String sinkTopic="dwm_unique_visit";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(sourceTopic, groupId);
        DataStreamSource<String> jsonStrDS = env.addSource(kafkaSource);


        //把读到的数据进行格式转换
        SingleOutputStreamOperator<JSONObject> jsonObjDS = jsonStrDS.map(jsonStr -> JSON.parseObject(jsonStr));

        //进行分组
        KeyedStream<JSONObject, String> keybyWithMidDS = jsonObjDS.keyBy(
                jsonObj -> jsonObj.getJSONObject("common").getString("mid")
        );

        //过滤
        SingleOutputStreamOperator<JSONObject> filteredDS = keybyWithMidDS.filter(
                new RichFilterFunction<JSONObject>() {

                    ValueState<String> lastVisitDateState = null;
                    SimpleDateFormat sdf = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyyMMdd");

                        ValueStateDescriptor<String> lastVisitDateStateDes =
                                new ValueStateDescriptor<>("lastVisitDateState", String.class);
                        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.days(1)).build();
                        lastVisitDateStateDes.enableTimeToLive(stateTtlConfig);
                        this.lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDes);
                    }

                    @Override
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        //判断当前页面是否从别的页面跳转过来的
                        String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        if (lastPageId != null && lastPageId.length() > 0) {
                            return false;
                        }

                        Long ts = jsonObj.getLong("ts");

                        //当前访问时间戳转换为日期字符串
                        String logDate = sdf.format(new Date(ts));
                        String lastVisitDate = lastVisitDateState.value();

                        //当前页面的访问时间和状态时间进行对比
                        if (lastVisitDate != null && lastVisitDate.length() > 0 && lastVisitDate.equals(logDate)) {
                            System.out.println("已访问：lastVisitDate-" + lastVisitDate + ",||logDate:" + logDate);
                            return false;
                        } else {
                            System.out.println("未访问：lastVisitDate-" + lastVisitDate + ",||logDate:" + logDate);
                            lastVisitDateState.update(logDate);
                            return true;
                        }
                    }
                }
        );

        //写会到kafak,json转为string
        SingleOutputStreamOperator<String> kafkaDS = filteredDS.map(jsonObj -> jsonObj.toJSONString());

        //写会到kafka dwm层
        kafkaDS.addSink(MyKafkaUtil.getKafkaSink(sinkTopic));

        env.execute();

    }
}
